package com.gitee.ixtf.vertx

import cn.hutool.core.compress.ZipWriter
import cn.hutool.core.util.RandomUtil
import cn.hutool.log.Log
import io.vertx.core.buffer.Buffer
import io.vertx.core.streams.ReadStream
import io.vertx.ext.reactivestreams.ReactiveReadStream
import java.io.InputStream
import java.io.OutputStream
import java.nio.charset.StandardCharsets.UTF_8
import java.util.concurrent.TimeUnit
import kotlinx.coroutines.reactor.flux
import kotlinx.coroutines.reactor.mono
import reactor.core.publisher.Sinks
import reactor.core.publisher.Sinks.EmitResult

class InputStreamState(
    bufferSize: Int = 512,
    rrs: ReactiveReadStream<Buffer> = ReactiveReadStream.readStream(),
    block: suspend () -> InputStream,
) : ReadStream<Buffer> by rrs {
  private val log by lazy { Log.get(this::class.java) }

  init {
    flux {
          block().use { inputStream ->
            val bufferArray = ByteArray(bufferSize)
            var bytes = inputStream.read(bufferArray)
            while (bytes >= 0) {
              val data = Buffer.buffer(bytes)
              data.setBytes(0, bufferArray, 0, bytes)
              send(data)
              bytes = inputStream.read(bufferArray)
            }
          }
        }
        .doOnError { log.error(it) }
        .subscribe(rrs)
  }
}

class ZipOutputStreamState(
    bufferSize: Int = 512,
    rrs: ReactiveReadStream<Buffer> = ReactiveReadStream.readStream(),
    resources: suspend () -> Iterable<Pair<String, suspend () -> InputStream>>,
) : ReadStream<Buffer> by rrs {
  private val log by lazy { Log.get(this::class.java) }
  private val sinkMany = Sinks.many().unicast().onBackpressureBuffer<Buffer>()
  private val bufferArray = ByteArray(bufferSize)
  private var bytes: Int = 0

  private inner class ConsumeOutputStream : OutputStream() {
    override fun write(b: Int) {
      if (bufferArray.size == bytes) {
        emitNext(Buffer.buffer(bufferArray))
        bytes = 0
      }
      bufferArray[bytes] = b.toByte()
      bytes += 1
    }

    override fun close() {
      if (bytes > 0) {
        val data = Buffer.buffer(bytes)
        data.setBytes(0, bufferArray, 0, bytes)
        emitNext(data)
        bytes = 0
      }
    }
  }

  private fun emitNext(data: Buffer, retry: Int = 5) {
    val failure = sinkMany.tryEmitNext(data).takeIf(EmitResult::isFailure) ?: return
    if (EmitResult.FAIL_OVERFLOW == failure) {
      val nextRetry = retry - 1
      if (nextRetry > 0) {
        TimeUnit.MILLISECONDS.sleep(RandomUtil.randomLong(500, 3000))
        return emitNext(data, nextRetry)
      }
    }
    failure.orThrow()
  }

  init {
    sinkMany.asFlux().subscribe(rrs)
    mono {
          ZipWriter(ConsumeOutputStream(), UTF_8).use { w ->
            resources().forEach { (p, b) -> b().use { w.add(p, it) } }
          }
        }
        .doOnSuccess {
          sinkMany.tryEmitComplete().takeIf(EmitResult::isFailure)?.let { log.error("$it") }
        }
        .doOnError {
          log.error(it)
          sinkMany.tryEmitError(it).takeIf(EmitResult::isFailure)?.let { log.error("$it") }
        }
        .subscribe()
  }
}
